package com.ipinyou.mprofile

import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{ Seconds, StreamingContext }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.streaming.Minutes

/**
 * Created by yuwenbing on 15/12/1.
 */

object kafkaSparkNCluster {

  // Function to create and setup a new StreamingContext
  def functionToCreateContext(): StreamingContext = {
    val sc = new SparkContext(new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[4]"))
    val ssc = new StreamingContext(sc, Seconds(20))
    ssc.checkpoint("file:///Users/miaoyujia/git/mobile-userprofile-src/mprofile/file/checkpointing") // set checkpoint directory
    
    
    ssc
  }

  def main(args: Array[String]): Unit = {

    // Get StreamingContext from checkpoint data or create a new one
    val ssc = StreamingContext.getOrCreate("file:///Users/miaoyujia/git/mobile-userprofile-src/mprofile/file/checkpointing", functionToCreateContext _)

    val brokers = "192.168.145.216:9092,192.168.145.217:9092,192.168.145.218:9092,192.168.145.221:9092,192.168.145.222:9092,192.168.145.223:9092"

    println("-----------------" + brokers)

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)

    val topicSet = Set("bid")

    val bidRdd = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicSet)

    val bid_RDD = bidRdd.map(_._2).flatMap { line =>

      //println(line)
      //val segs = line.split("\t")
      // 新日志 格式修改
      //   区间
      val segs = line.split("\u0001", -1)

      if (segs.length > 10) {

        // 字段
        val puser = segs(2)
        val arruser = puser.split("\u0002", -1)

        val papp = segs(5)
        val arrapp = papp.split("\u0002", -1)

        var pyid = "error"
        var appid = "error"

        if (arruser.length >= 6) {
          pyid = arruser(0)
        }

        if (arrapp.length > 10) {
          appid = "appid" + arrapp(1)
        }
        List((appid, pyid))
      } else {
        List(("error", "error"))
      }
    }.filter(x => (!x._1.contains("error")) || (!x._2.contains("error")))

    bid_RDD.count().print()

    ssc.start()
    ssc.awaitTermination()

  }

}
